package com.leigq.www.consumer.consumer;

import com.leigq.www.common.constant.QueueConstant;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * 订单消费客户端
 * <br/>
 * 延迟队列参考：<a href='https://blog.csdn.net/wangpf2011/article/details/86600809'>分布式架构核心组件之消息队列RabbitMQ</a>
 * @author leiguoqing
 */
@Component
@Slf4j
public class OrderConsumer {

	@Value("${spring.rabbitmq.listener.simple.retry.max-attempts}")
    private Integer maxAttempts;

	/*↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ JavaConfig 方式实现延迟队列，延迟队列的创建及绑定关系已在生成端配置，这里只需绑定实际处理队列即可↓↓↓↓↓↓↓↓*/
	/**
	 * 处理方式一普通队列延时后的消费队列（实际消费队列）
	 */
	@RabbitListener(queues = QueueConstant.DELAY_DLX_QUEUE1_NAME)
	@RabbitHandler
	public void delayDlxQueue1Consumer(@Headers Map<String, Object> headers, Channel channel, Message message) throws IOException {
		String order = new String(message.getBody(), StandardCharsets.UTF_8);

		/*
		 * Delivery Tag 用来标识信道中投递的消息。RabbitMQ 推送消息给 Consumer 时，会附带一个 Delivery Tag
		 * 以便 Consumer 可以在消息确认时告诉 RabbitMQ 到底是哪条消息被确认了。
		 * RabbitMQ 保证在每个信道中，每条消息的 Delivery Tag 从 1 开始递增。
		 * */
		Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);

		/*
		 * 手动签收 ACK, 确认一条消息已经被消费
		 * multiple 取值为 false 时，表示通知 RabbitMQ 当前消息被确认
		 * 如果为 true，则额外将比第一个参数指定的 delivery tag 小的消息一并确认
		 * */
		channel.basicAck(deliveryTag, false);

		log.info("处理方式一普通队列延时后的消费队列（实际消费队列）成功: order = [{}]", order);
	}

	/**
	 * 处理方式二延时队列
	 */
	@RabbitListener(queues = QueueConstant.DELAY_QUEUE2_NAME)
	@RabbitHandler
	public void delayQueue2Consumer(@Headers Map<String, Object> headers, Channel channel, Message message) throws IOException {
		String order = new String(message.getBody(), StandardCharsets.UTF_8);

		/*
		 * Delivery Tag 用来标识信道中投递的消息。RabbitMQ 推送消息给 Consumer 时，会附带一个 Delivery Tag
		 * 以便 Consumer 可以在消息确认时告诉 RabbitMQ 到底是哪条消息被确认了。
		 * RabbitMQ 保证在每个信道中，每条消息的 Delivery Tag 从 1 开始递增。
		 * */
		Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);

		/*
		 * 手动签收 ACK, 确认一条消息已经被消费
		 * multiple 取值为 false 时，表示通知 RabbitMQ 当前消息被确认
		 * 如果为 true，则额外将比第一个参数指定的 delivery tag 小的消息一并确认
		 * */
		channel.basicAck(deliveryTag, false);
		log.info("处理方式二延时队列成功：order = [{}]", order);
	}
	/*↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ JavaConfig 方式实现延迟队列 ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑*/


	/*↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ 全注解方式 实现消息队列（非延迟） ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓*/
	/**
	 * 订单消费队列 1
	 * <br/>
	 * 俩个 Customer 订阅同一个队列，队列中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。
     * <br/>
     * <code>@RabbitListener</code>: 配置监听的哪一个队列，同时在没有 queue 和 exchange 的情况下会去创建并建立绑定关系
	 */
	@RabbitListener(
			bindings = @QueueBinding(
					value = @Queue(value = QueueConstant.ORDER_QUEUE_NAME, durable = "true", autoDelete = "false"),
					exchange = @Exchange(name = QueueConstant.ORDER_QUEUE_EXCHANGE_NAME, type = "topic"),
					key = QueueConstant.ORDER_QUEUE_ROUTING_KEY
			),
			// 并发
            concurrency = "5"
	)
	@RabbitHandler
	public void onOrderMessage1(@Headers Map<String, Object> headers, Channel channel, Message message) throws IOException {
		this.onOrderMessage2(headers, channel, message);
	}

	/**
	 * 订单消费队列 2
	 * <br/>
	 * 俩个 Customer 订阅同一个队列，队列中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。
     * <br/>
     * <code>@RabbitListener</code>: 配置监听的哪一个队列，同时在没有 queue 和 exchange 的情况下会去创建并建立绑定关系
	 */
	@RabbitListener(
			bindings = @QueueBinding(
					value = @Queue(value = QueueConstant.ORDER_QUEUE_NAME, durable = "true"),
					exchange = @Exchange(name = QueueConstant.ORDER_QUEUE_EXCHANGE_NAME, type = "topic"),
					key = QueueConstant.ORDER_QUEUE_ROUTING_KEY
			)
	)
	@RabbitHandler
	public void onOrderMessage2(@Headers Map<String, Object> headers, Channel channel, Message message) throws IOException {
		String order = new String(message.getBody(), StandardCharsets.UTF_8);

		// 消费者操作
        log.warn("onOrderMessage2 收到消息: order = [{}]，开始消费...", order);

        log.info("headers: [{}]", headers);
        log.info("channel: [{}]", channel);

        /*
         * Delivery Tag 用来标识信道中投递的消息。RabbitMQ 推送消息给 Consumer 时，会附带一个 Delivery Tag
         * 以便 Consumer 可以在消息确认时告诉 RabbitMQ 到底是哪条消息被确认了。
         * RabbitMQ 保证在每个信道中，每条消息的 Delivery Tag 从 1 开始递增。
         * */
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);

        try {
            // 模拟异常，放开注释即可正常消费
            // int a = 1 / 0;

            /*
             * 这里实现实体的业务逻辑，多个消费者处理是需注意并发问题（加锁）
             * */
            log.warn("onOrderMessage2 执行了相关业务....");

            log.warn("onOrderMessage2 消费完成....");

            /*
             * 手动签收 ACK, 确认一条消息已经被消费
             * multiple 取值为 false 时，表示通知 RabbitMQ 当前消息被确认
             * 如果为 true，则额外将比第一个参数指定的 delivery tag 小的消息一并确认
             * */
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            // 判断重试次数
            int retries = Integer.parseInt(headers.get("x-retries") + "") + 1;
            log.info("消费异常，重试次数：{}", retries);
            final MessageProperties messageProperties = message.getMessageProperties();
            messageProperties.setHeader("x-retries", retries);

            if (retries >= maxAttempts) {
                /*
                 * 手动签收 ACK, 确认一条消息已经被消费
                 * multiple 取值为 false 时，表示通知 RabbitMQ 当前消息被确认
                 * 如果为 true，则额外将比第一个参数指定的 delivery tag 小的消息一并确认
                 * */
                channel.basicAck(deliveryTag, false);
                log.info("重试后手动签收ACK, deliveryTag:{}, order: {}", deliveryTag, order);
            }

			// 主动抛异常触发重试及投放到消费错误队列
            log.error("异常信息：", e);
            throw new RuntimeException("消费异常" );
        }
	}
	/*↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ 全注解方式 实现消息队列（非延迟） ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑*/


    /**
     * 消费错误队列
     * <br/>
     * <code>@RabbitListener</code>: 配置监听的哪一个队列，同时在没有 queue 和 exchange 的情况下会去创建并建立绑定关系
     */
    @RabbitListener(queues = {QueueConstant.ERROR_QUEUE_NAME})
    @RabbitHandler
    public void onErrorMessage(@Headers Map<String, Object> headers, Channel channel, Message message) throws IOException {
        String msg = new String(message.getBody(), StandardCharsets.UTF_8);
        log.info("处理消费失败消息:{}", msg);
        // 可将消费错误的队列进行持久化，做后续处理（在headers里面添加x-biz-type可进行业务区分）
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    }

}
